package com.chukun.multi.pipeline.reusable;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

public abstract class AbstractParallelPipe<IN,OUT,V> extends AbstractPipe<IN,OUT> {

    private final ExecutorService executorService;

    public AbstractParallelPipe(BlockingQueue<IN> queue,ExecutorService executorService) {
        super();
        this.executorService = executorService;
    }

    /**
     * 留给子类实现。用于根据指定的输入元素构造一组子任务
     * @param input
     * @return
     * @throws Exception
     */
    protected abstract List<Callable<V>> buildTasks(IN input) throws Exception;

    /**
     * 以并行的方式执行一组子任务
     * @param subTaskResults
     * @return
     * @throws Exception
     */
    protected abstract OUT combineResults(List<Future<V>> subTaskResults) throws Exception;

    /**
     * 以并行的方式执行一组任务
     * @param tasks
     * @return
     * @throws Exception
     */
    protected List<Future<V>> invokeParallel(List<Callable<V>> tasks) throws Exception {
        return executorService.invokeAll(tasks);
    }

    /**
     *
     * @param input
     * @return
     */
    protected OUT doProcess(final IN input) throws PipeException {
        OUT out = null;
        try{
            out = combineResults(invokeParallel(buildTasks(input)));
        }catch (Exception e) {
            throw  new PipeException(this,input,"task failed",e);
        }
        return out;
    }
}
